/*
 * RMTransactionManager.java   0.0.1    9 Nov 2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.transactions.management.rm;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.mentalsmash.documentation.Concurrency;
import org.mentalsmash.tazio.cache.ServerCache;
import org.mentalsmash.tazio.commons.identifiers.OID;
import org.mentalsmash.tazio.commons.timer.MonitorTimePerformances;
import org.mentalsmash.tazio.transactions.RollbackException;
import org.mentalsmash.tazio.transactions.TransactionStatus;
import org.mentalsmash.tazio.transactions.TransactionalSystemException;
import org.mentalsmash.tazio.transactions.log.TransactionsLog;
import org.mentalsmash.tazio.transactions.management.CommitmentAction;
import org.mentalsmash.tazio.transactions.management.TazioTransaction;
import org.mentalsmash.tazio.transactions.management.TransactionStateException;
import org.mentalsmash.tazio.transactions.management.ValidationAgent;
import org.mentalsmash.tazio.transactions.modules.TxQueue;
import org.mentalsmash.tazio.transactions.objects.RootSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.internal.Preconditions;

/**
 * @author Andrea Reale
 * @version 0.0.1
 */
@Concurrency(concurrentAccess = true, specification = "the RMTransactionManager being a Singeleton, is inherently accessed by "
	+ "multiple threads at a time. Nevertheless it has just one non final "
	+ "status variable, that is the boolean var that tells if the commitment "
	+ "section is free or not. Accesses to that variable are synchronized "
	+ "using the commitmentLock")
@Singleton
class RMTransactionManagerImpl implements RMTransactionManager {
    private final static Logger log = LoggerFactory
	    .getLogger(RMTransactionManagerImpl.class);

    private final static String COMMITTER_NAME = "RM-Committer-Thread";
    // RMTm's name
    private final OID name;

    // instance of TazioTransaction local to a thread
    // private final ThreadLocal<TazioTransaction> currentTransaction = new
    // ThreadLocal<TazioTransaction>();

    private final Lock counterLock = new ReentrantLock();
    private long seqNoCounter = 0;

    private final TransactionsLog transactionsLog;

    private final RootSet rootSet;

    private final ValidationAgent validationAgent;

    // One thread at a time may be in the commitment section. The lock ensures
    // mutual exclusion of threads
    private final Lock commitmentLock = new ReentrantLock();

    private final BlockingQueue<TazioTransaction> awaitingCommit;
    
    private final ServerCache cache;

    @Inject
    RMTransactionManagerImpl(final RootSet rs, final TransactionsLog txLog,
	    final ValidationAgent vAgent, @TxQueue final int txQueueSize, final ServerCache cache) {

	this.name = OID.generateOID();

	this.rootSet = rs;
	this.transactionsLog = txLog;
	this.validationAgent = vAgent;
	
	this.cache = cache;

	this.awaitingCommit = new ArrayBlockingQueue<TazioTransaction>(
		txQueueSize, true);

	Thread committerThread = new Thread(new Committer(this, awaitingCommit));
	committerThread.setName(COMMITTER_NAME);
	committerThread.start();
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transactions.SystemTransactionManager#commit()
     */
    @MonitorTimePerformances
    protected void commit(final TazioTransaction tx) throws RollbackException,
	    TransactionStateException, TransactionalSystemException {

	Preconditions.checkNotNull(tx);
	Preconditions
		.checkArgument(tx.getStatus() == TransactionStatus.GLOBAL_VALIDATION);

	final CommitmentAction action = new RMCommitmentAction(tx, this,
		this.rootSet, this.cache);

	try {
	    getCommitmentLock().lock();
	    try {
		log.debug("IN HERE");
		action.call();
	    } finally {
		getCommitmentLock().unlock();
	    }

	} catch (final TransactionStateException e) {
	    throw e;
	} catch (final RollbackException e) {
	    throw e;
	} catch (final Exception e) {
	    throw new TransactionalSystemException(e);
	}

    }

    // /**
    // * Associates the given transaction to the thread calling this method.
    // *
    // * @param tx
    // * the tx to associate to the calling method
    // */
    // protected void setThreadsTransaction(TazioTransaction tx) {
    // Preconditions.checkNotNull(tx);
    //
    // this.currentTransaction.set(tx);
    // }

    // /*
    // * (non-Javadoc)
    // *
    // * @see
    // *
    // org.mentalsmash.tazio.transactions.SystemTransactionManager#getTransaction
    // * ()
    // */
    // @Override
    // public Transaction getTransaction() {
    // final TazioTransaction tx = currentTransaction.get();
    // if (tx == null) {
    // return NullTazioTransaction.getInstance();
    // } else {
    // return tx;
    // }
    // }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.SystemTransactionManager#
     * getTransactionsLog()
     */
    @Override
    public TransactionsLog getTransactionsLog() {
	return this.transactionsLog;
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.SystemTransactionManager#
     * getValidationAgent()
     */
    @Override
    public ValidationAgent getValidationAgent() {
	return this.validationAgent;
    }

    private Lock getCommitmentLock() {
	return this.commitmentLock;
    }

    @Override
    public long incrementAndGetSeqNoCtr() {
	lockCounter();
	try {
	    this.seqNoCounter++;
	    return this.seqNoCounter;
	} finally {
	    unlockCounter();
	}

    }

    /**
     * Returns the value of the sequence number counter of this RM
     * 
     * @return the value of the sequence number counter of this RM
     */
    @Override
    public long getSeqNoCtr() {
	lockCounter();
	try {
	    return this.seqNoCounter;
	} finally {
	    unlockCounter();
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.management.SystemTransactionManager
     * #getManagerName()
     */
    @Override
    public OID getManagerName() {
	return this.name;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.management.rm.RMTransactionManager
     * #offerTransactionForCommit
     * (org.mentalsmash.tazio.transactions.management.TazioTransaction)
     */
    @Override
    public boolean offerTransactionForCommit(final TazioTransaction tx) {
	boolean accepted = false;
	try {
	    accepted = this.awaitingCommit.offer(tx,1000,TimeUnit.MILLISECONDS);
	} catch (InterruptedException e) {
	    e.printStackTrace();
	}
	if(accepted) {
	    log.debug("Accepted transaction {}", tx.getTransactionId());
	} else {
	    log.debug("Refused transaction {}", tx.getTransactionId());
	}
	return accepted;

    }

    @Override
    public void lockCounter() {
	counterLock.lock();
	//log.debug("COUNTER LOCKED");

    }

    @Override
    public void unlockCounter() {
	counterLock.unlock();
	//log.debug("COUNTER UNLOCKED");
    }

}

class Committer implements Runnable {

    private static Logger log = LoggerFactory.getLogger(Committer.class);
    private volatile boolean running;
    private final BlockingQueue<TazioTransaction> txQueue;
    private final RMTransactionManagerImpl tm;

    protected Committer(final RMTransactionManagerImpl tm,
	    final BlockingQueue<TazioTransaction> txQueue) {
	Preconditions.checkNotNull(txQueue);
	this.running = false;
	this.tm = tm;
	this.txQueue = txQueue;
    }

    /*
     * (non-Javadoc)
     * 
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
	Preconditions.checkState(!running,
		"Cannot run, committer already running");

	running = true;
	while (running) {
	    TazioTransaction toCommit = null;
	    try {
		toCommit = txQueue.take();
	    } catch (final InterruptedException e) {
		log.warn("Committer was interrupted, restarting. Cause: ", e);
		continue;
	    }
	    
	    log.debug("Committing tranaction {}", toCommit.getTransactionId());

	    try {
		tm.commit(toCommit);
	    } catch (final RollbackException e) {
		log.debug("Transaction {} rolled back", toCommit
			.getTransactionId());
		continue;
	    } catch (final Exception e) {
		log.warn("Unexpected exception while committing a transaction",
			e);
		continue;
	    }
	}
    }

    public void stop() {
	log.info("Stopping the committer");
	running = false;
    }

}